package cn.linjianhui.demo.kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.Random;

public class ProducerTest {
    public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.8.251:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);

        Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());

        for (int i = 0; i < 10; i++) {
            int value = new Random().nextInt(100);
            System.out.println(String.format("%s:%d", Integer.toString(i), value));
            producer.send(new ProducerRecord<>("OMS-Order", Integer.toString(i), Integer.toString(value)), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if(e != null) {
                        e.printStackTrace();
                    } else {
                        System.out.println("The offset of the record we just sent is: " + recordMetadata.offset());
                    }
                }
            });
            Thread.sleep(1000);
        }

        System.out.println("生产数据成功");
        producer.close();
    }
}
